package start.spring.basic.logs.service.logsrecord;

import java.util.ArrayList;
import java.util.List;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.jboss.logging.MDC;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import com.alibaba.fastjson.JSONObject;

import start.spring.basic.log.LogConstant;
import start.spring.basic.logs.service.BaseServiceImpl;
import start.spring.basic.service.IKfkConsumerHandleService;
import start.spring.basic.util.common.StringUtil;

/**
 * 
 * ****************************************************************************
 * kafka中的日志记录到数据库表
 *
 * @author(作者)：xuyongyun	
 * @date(创建日期)：2023年2月15日
 ******************************************************************************
 */
@Service("recordKfkLogsToDb")
public class LogsConsumerServiceImpl extends BaseServiceImpl implements ILogsConsumerService,IKfkConsumerHandleService {
	
	@Transactional
	@Override
	public void handle(ConsumerRecords<String,String> records) throws Exception {
		
		if(records!=null&&!records.isEmpty()) {
			//设置链路traceId
    		String traceId = StringUtil.getUuid();
    		MDC.put(LogConstant.REQUEST_KEY_TRACEID, traceId);
    		
			String sql = "INSERT INTO gg_sys_log (biz_type, user_id, user_name, class_name, "
 					   + "method_name, code, msg, sys_code, url, start_time, end_time, clhs, "
 					   + "request_args, oper_result, oper_detail, trace_id) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
 			List<Object[]> params = new ArrayList<>();
 			for(ConsumerRecord<String,String> msg : records) {
 				JSONObject logInfo = JSONObject.parseObject(msg.value());
 				String startTime = StringUtil.isEmpty(logInfo.getString("startTime"))?null:logInfo.getString("startTime");
 				String endTime = StringUtil.isEmpty(logInfo.getString("endTime"))?null:logInfo.getString("endTime");
 				String clhs = StringUtil.isEmpty(logInfo.getString("clhs"))?null:logInfo.getString("clhs");
 				String[] param = {
 						          logInfo.getString("businessType"),logInfo.getString("userId"),
 								  logInfo.getString("userName"),logInfo.getString("className"),
 								  logInfo.getString("methodName"),logInfo.getString("code"),
 								  logInfo.getString("msg"),logInfo.getString("systemType"),
 								  logInfo.getString("url"),startTime,endTime,clhs,
 								  logInfo.getString("requestArgs"),logInfo.getString("operResult"),
 								  logInfo.getString("operDetail"),logInfo.getString("traceId")
 								 };
 				params.add(param);
 			}
 			super.dao.batchExcute(sql, params);
			
			MDC.clear();
		}
	}

}
